Kaggle Submission_Final(Python)

import nltk
import re
from nltk.corpus import stopwords
from nltk.sentiment.vader import SentimentIntensityAnalyzer
from pyspark.sql.functions import *
from nltk.stem import WordNetLemmatizer
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.classification import LogisticRegression, RandomForestClassifier, GBTClassifier
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.sql import SparkSession
from sparknlp.annotator import Tokenizer, Normalizer, StopWordsCleaner, Stemmer
from pyspark.sql.types import *
from pyspark.sql.functions import *
import matplotlib.pyplot as plt
spark.conf.set("spark.sql.legacy.timeParserPolicy", "LEGACY")
from pyspark.ml import Pipeline
from sklearn.feature_extraction.text import TfidfVectorizer
from pyspark.ml.feature import RegexTokenizer, StopWordsRemover, CountVectorizer, IDF, NGram, VectorAssembler, Word2Vec
from pyspark.sql.window import Window
from pyspark.ml.feature import *
from nltk.stem import SnowballStemmer

# nltk.download('all')

# Load in one of the tables
df = spark.sql("select * from default.reviews_train").sample(0.25, seed=47)
print((df.count(), len(df.columns)))

df.printSchema()

# Let's look at some quick summary statistics
display(df.withColumnRenamed('summary', 'summaryCol').describe())

# The count of each overall rating
display(df.groupBy("overall").count().orderBy(col("overall").asc()))

# The most common product IDs
display(df.groupBy("asin").count().orderBy(col("count").desc()))

def date_time_extraction(df):

    # Assuming df is your DataFrame and reviewTime is the column name
    df = df.withColumn("reviewDate", to_date(df.reviewTime, "MM dd, yyyy"))

    # Extract the month, day, and year into new columns
    df = df.withColumn("reviewMonth", month("reviewDate"))
    df = df.withColumn("reviewDay", dayofmonth("reviewDate"))
    df = df.withColumn("reviewYear", year("reviewDate"))
    df = df.withColumn('reviewdayOfWeek', dayofweek(col('reviewDate')))
    df = df.withColumn("reviewYearDate", to_date(col("reviewYear"), "yyyy"))
    df = df.withColumn("verified", when(col("verified") == True, 1).otherwise(0))

    return df

df = date_time_extraction(df)

def review_counts_per_customer(data_input):
    
    # Finding the total number of reviews a customer left in their lifetime and the unique products reviewed

    review_counts = data_input.groupBy("reviewerID").agg(
        countDistinct('reviewID').alias('reviews_per_customer'),
        countDistinct('asin').alias('products_reviewed_per_customer'))
    
    data_input = data_input.join(review_counts, on="reviewerID", how="inner")

    return data_input

df = review_counts_per_customer(df)

def review_counts_per_product(data_input):
    
    # Finding the total number of reviews a product generated in its lifetime

    review_counts = data_input.groupBy("asin").agg(
        countDistinct("reviewID").alias("reviews_per_product"),
        countDistinct('reviewerID').alias("reviewers_per_product"),
        min('reviewDate').alias('product_earliest_review'),
        max('reviewDate').alias('product_latest_review'))
        
    data_input = data_input.join(review_counts, on="asin", how="inner")
    
    data_input = data_input.withColumn('product_review_interval', datediff(data_input['product_latest_review'], data_input['product_earliest_review']))

    data_input = data_input.withColumn("product_review_interval", when(col("product_latest_review") == col("product_earliest_review"), 0).otherwise(col("product_review_interval")))
    
    data_input = data_input.withColumn('product_earliest_review', datediff(current_date(), data_input['product_earliest_review']))
    
    data_input = data_input.withColumn('product_latest_review', datediff(current_date(), data_input['product_latest_review']))
        
    return data_input

df = review_counts_per_product(df)

display(df)

# Filtering out rows 

rows_with_missing = df.filter(col("reviewText").isNull() | col("summary").isNull() | col("label").isNull())
display(rows_with_missing)

# Find rows with empty strings in specified columns
empty_rows = df.filter(col("summary") == " ")

# Show the rows with empty strings
display(empty_rows)

df = df.filter(col("summary") != " ")
print((df.count(), len(df.columns)))

def sentiment_analysis(data_input):

    from pyspark.sql import SparkSession
    from pyspark.sql.functions import udf
    from pyspark.sql.types import StringType, StructType, StructField, FloatType
    from nltk.sentiment.vader import SentimentIntensityAnalyzer

    # Initialize the VADER sentiment analyzer
    analyzer = SentimentIntensityAnalyzer()

    # Define a UDF to perform sentiment analysis
    def analyze_sentiment(text):
        sentiment = analyzer.polarity_scores(text)
        return sentiment["compound"]

    # Register the UDF with FloatType return type
    sentiment_udf = udf(analyze_sentiment, FloatType())

    # Apply the sentiment analysis UDF to the DataFrame
    data_input = data_input.withColumn("sentiment_score", sentiment_udf(data_input["reviewText"]))

    # Show the result
    return data_input

def review_level_features(df):

    # Calculating the length of every review - negative reviews tend to be longer

    df = df.withColumn("Words", split(df["reviewText"], " "))

    df = df.withColumn("Review_Length", size(df["Words"]))

    # Bucketing reviews into high and low

    condition = (col("overall") > 2)

    df = df.withColumn("overall_high", when(condition, 1).otherwise(0))

    # Bucketing customer reviews into positive & negative / neutral experiences

    df = sentiment_analysis(df)

    sentiment_condition = (col("sentiment_score") > 0)

    df = df.withColumn("positive_experience", when(sentiment_condition, 1).otherwise(0))

    neutral_exp_condition = (col("sentiment_score") == 0)

    df = df.withColumn("neutral_experience", when(neutral_exp_condition, 1).otherwise(0))

    # Do customers review only if there is a problem? Or do they always review in general? 

    result = df.groupBy("reviewerID").agg(

        avg("sentiment_score").alias("average_sentiment_score_per_customer"),

        percentile_approx("sentiment_score", 0.5).alias("median_sentiment_score_per_customer"))

    negative_condition = (col("average_sentiment_score_per_customer") < 0)

    result = result.withColumn("negative_exp_reviewer", when(negative_condition, 1).otherwise(0))

    df = df.join(result, on="reviewerID", how="left")

    # General sentiment of a product

    sentiment_product = df.groupBy("asin").agg(

        avg("sentiment_score").alias("avg_sentiment_score_per_product"),

        percentile_approx("sentiment_score", 0.5).alias("median_sentiment_score_per_product")

    )

    df = df.join(sentiment_product, on="asin", how="left")

    df = df.withColumn("review_divergence", ((col("sentiment_score") - col("avg_sentiment_score_per_product")) / col("avg_sentiment_score_per_product")))

    df = df.withColumn("review_divergence", when(col("review_divergence").isNull(), 0).otherwise(col("review_divergence")))

    columns_to_drop = [

        # "average_sentiment_score_per_customer",

                       "sentiment_score"]

    df = df.drop(*columns_to_drop)

    return df

df = review_level_features(df)

def customer_level_date_time_features(df):

    # Calculate the number of days since review creation - idea is to find out how recent a review is. More recent reviews are more relevant than older ones
    
    df = df.withColumn("days_since_review_creation", datediff(current_date(), col("reviewDate")))

    df = df.withColumn("months_since_review_creation", col("days_since_review_creation") / 30)

    # Finding the first & last review of each reviewer
    
    result = df.groupBy("reviewerID").agg(
        min("reviewDate").alias("first_review_date"), 
        max("reviewDate").alias("last_review_date"))

    df = df.join(result, on="reviewerID", how="inner")

    # Finding the number of reviews the customer leaves per year

    df = df.withColumn("reviews_per_year", col("reviews_per_customer") / (datediff(col("last_review_date"), col("first_review_date")) / 365))

    # Shows how long the account was active for

    df = df.withColumn("months_since_customers_first_review", datediff(current_date(), col("first_review_date")) / 30)

    # Shows when the most recent review happened

    df = df.withColumn("months_since_customers_last_review", datediff(current_date(), col("last_review_date")) / 30)

    # df = df.withColumn("reviews_per_year", when(col("first_review_date") == col("last_review_date"), 1).otherwise(col("reviews_per_year")))

    df = df.withColumn("reviews_per_year_per_customer", when(col("first_review_date") == col("last_review_date"), 1).otherwise(col("reviews_per_year")))

    columns_to_drop = ["reviewYearDate",
                       "last_review_date",
                       "first_review_date", 
                       "reviews_per_year"
                    ]

    df = df.drop(*columns_to_drop)

    return df

df = customer_level_date_time_features(df)

agg_df = df.filter(col("label") == 1).groupBy("reviewMonth").agg(count('*').alias("review_count"))

# Step 4: Convert the PySpark DataFrame to a Pandas DataFrame
pandas_df = agg_df.toPandas()

# Step 5: Create the time series graph using Matplotlib
plt.figure(figsize=(12, 6))
plt.bar(pandas_df["reviewMonth"], pandas_df['review_count'], color='b')
plt.title('Customer Reviews Over Time')
plt.xlabel('Time')
plt.ylabel('Review Count')
plt.grid(True)
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

def org_level_features(df):

    condition = ((col("reviewMonth") >= 10) | (col("reviewMonth") == 1))
    df = df.withColumn("High_Volume_Timeframes", when(condition, 1).otherwise(0))

    columns_to_drop = ["reviewDate", 
                       "reviewMonth", "reviewDay", "reviewYear"
                    ]

    df = df.drop(*columns_to_drop)
    
    return df

df = org_level_features(df)

df.columns

def cleaning_review_data(df):

    stemmer = SnowballStemmer(language="english")

    def stem_text(text):
        return " ".join([stemmer.stem(word) for word in text.split()])

    stem_udf = udf(stem_text, StringType())

    stemmed_df = df.withColumn("stemmed_text", stem_udf(col("reviewText")))

    # convert text to all lower cases
    lower_df = stemmed_df.withColumn("lowered_text", lower(col("stemmed_text")))

    # remove punctuations
    cleaned_df = lower_df.withColumn("cleaned_text", regexp_replace(col("lowered_text"), "[^a-zA-Z\\s]", ""))

    # tokenize reviewText in order to filter out stop words
    tokenizer = Tokenizer(inputCol="cleaned_text", outputCol="words")
    tokenized_df = tokenizer.transform(cleaned_df)

    remover = StopWordsRemover(inputCol="words", outputCol="filtered_text")
    filtered_df = remover.transform(tokenized_df)

    df = filtered_df.withColumn('simplified_reviewText', concat_ws(' ', 'filtered_text'))

    df = df.withColumn('review_length_original_chars', length(df['reviewText']))
    df = df.withColumn('review_length_simplified_chars', length(df['simplified_reviewText']))

    return df

def detecting_non_verbal_cues(df):
    
    # UDF to count the number of question marks
    def count_question_marks(text):
        return text.count("?")

    # UDF to count the number of exclamation marks
    def count_exclamation_marks(text):
        return text.count("!")

    # UDF to count the number of all-capital words
    def count_all_caps(text):
        return len([word for word in text.split() if word.isupper()])

    # Register UDFs
    count_question_marks_udf = udf(count_question_marks, IntegerType())
    count_exclamation_marks_udf = udf(count_exclamation_marks, IntegerType())
    count_all_caps_udf = udf(count_all_caps, IntegerType())

    # Apply UDFs to DataFrame
    df = df.withColumn("question_marks_count", 
                       count_question_marks_udf("reviewText")) \
                        .withColumn("exclamation_marks_count", count_exclamation_marks_udf("reviewText")) \
                        .withColumn("all_caps_count", count_all_caps_udf("reviewText"))

    return df

df = detecting_non_verbal_cues(df)
df = cleaning_review_data(df)

drop_list = [
    'asin', 
    'reviewID', 
    'reviewerID', 
    'unixReviewTime',
    'reviewTime', 
    'image', 
    'style', 
    'reviewerName', 
    'overall', 
    'stemmed_text', 
    'lowered_text', 
    'cleaned_text', 
    'words', 
    'filtered_text']

df = df.select([column for column in df.columns if column not in drop_list])
print((df.count(), len(df.columns)))

display(df)

df = df.na.drop(subset=df.columns)
# display(df)
print((df.count(), len(df.columns)))

df.printSchema()

def freq_word_count(data):

    regexTokenizer = RegexTokenizer(inputCol="reviewText", outputCol="words", pattern="\\W")
    data = regexTokenizer.transform(data)

    data = data.withColumn("wordCount", size(data["words"]))

    word_counts_label0 = data.filter(col("label") == 0).select(explode(col("words")).alias("word")).groupBy("word").agg(count("*").alias("count_label0"))

    word_counts_label1 = data.filter(col("label") == 1).select(explode(col("words")).alias("word")).groupBy("word").agg(count("*").alias("count_label1"))

    word_counts = word_counts_label0.join(word_counts_label1, "word", "outer").fillna(0)

    word_counts = word_counts.withColumn("frequent_compare", col("count_label1") / col("count_label0"))
    filtered_words = word_counts.filter((col("frequent_compare") > 2) & (col("count_label1") > 10))

    freq_word_list = [row.word for row in filtered_words.select("word").collect()]
    freq_word_set = set(freq_word_list)

    def count_freq_words(words):
        return len([word for word in words if word in freq_word_set])

    count_freq_words_udf = udf(count_freq_words, IntegerType())
    data = data.withColumn("freq_word_count", count_freq_words_udf(col("words")))

    return data, filtered_words

df, filtered_words = freq_word_count(df)

display(df)

# set seed for reproducibility
(trainingData, testingData) = df.randomSplit([0.8, 0.2], seed = 47)
print("Training Dataset Count: " + str(trainingData.count()))
print("Test Dataset Count: " + str(testingData.count()))

feature_names = ['featuresCount',             
                                            'countVectorsSum', 
                                            'freq_word_count',
                                            'verified',
                                            'reviews_per_customer',
                                            'products_reviewed_per_customer',
                                            'reviews_per_product',
                                            'Review_Length',
                                            'overall_high',
                                            'positive_experience',
                                            'neutral_experience',
                                            'negative_exp_reviewer',
                                            'days_since_review_creation',
                                            'months_since_review_creation',
                                            'months_since_customers_first_review',
                                            'months_since_customers_last_review',
                                            'reviews_per_year_per_customer',
                                            'High_Volume_Timeframes',
                                            'reviewdayOfWeek',
                                            'reviewers_per_product',
                                            'product_earliest_review',
                                            'product_latest_review',
                                            'review_length_original_chars',
                                            'review_length_simplified_chars',
                                            'avg_sentiment_score_per_product',
                                            'review_divergence',
                                            'average_sentiment_score_per_customer',
                                            'median_sentiment_score_per_customer',
                                            'median_sentiment_score_per_product',
                                            # 'average_sentiment_score_per_product',
                                            'tfidf',
                                            'question_marks_count',
                                            'exclamation_marks_count',
                                            'all_caps_count'
                                            ]

# We'll tokenize the text using a simple RegexTokenizer
regexTokenizer = RegexTokenizer(inputCol="reviewText", outputCol="words_1", pattern="\\W")
regexTokenizerSum = RegexTokenizer(inputCol="summary", outputCol="wordsSum", pattern="\\W")

# Remove standard Stopwords
stopwordsRemover = StopWordsRemover(inputCol="words_1", outputCol="filtered")
stopwordsRemoverSum = StopWordsRemover(inputCol="wordsSum", outputCol="filteredSum")

countVectors = CountVectorizer(inputCol="filtered", outputCol="featuresCount", vocabSize=3000, minTF=1, maxDF=0.40)

countVectorsSum = CountVectorizer(inputCol="filteredSum", outputCol="countVectorsSum", vocabSize=3000, minTF=1, maxDF=0.40)

# Generate Inverse Document Frequency weighting
# idf = IDF(inputCol="featuresCount", outputCol="idfFeatures", minDocFreq=100)

hashing_tf = HashingTF(inputCol="words_1", outputCol="raw_features_hash", numFeatures=50)
tfidf = IDF(inputCol="raw_features_hash", outputCol="tfidf")

featureAssembler = VectorAssembler(inputCols=feature_names, outputCol="features")

# ml_alg  = LogisticRegression(maxIter=10, regParam=0.3, elasticNetParam=0.0)

pipeline = Pipeline(stages=[regexTokenizer, regexTokenizerSum,
                            stopwordsRemover, 
                            countVectors, 
                            stopwordsRemoverSum, 
                            countVectorsSum,
                            hashing_tf,
                            tfidf,
                            featureAssembler,
                            ])

# paramGrid = ParamGridBuilder() \
#     .addGrid(countVectors.minTF, [1, 3, 5, 7, 10, 25, 50]) \
#     .addGrid(countVectors.maxDF, [0.1, 0.2, 0.3, 0.4, 0.5, 0.6]) \
#     .addGrid(countVectors.vocabSize, [500, 1000, 2000, 2500, 3000, 5000]) \
#     .build()

# Fit the pipeline to training documents.

pipelineFit = pipeline.fit(trainingData)
trainingDataTransformed = pipelineFit.transform(trainingData)
# display(trainingDataTransformed)

lr = LogisticRegression(maxIter=20, regParam=0.3, elasticNetParam=0)

lrModel = lr.fit(trainingDataTransformed)
trainingSummary = lrModel.summary

print("Training Accuracy:  " + str(trainingSummary.accuracy))
print("Training Precision: " + str(trainingSummary.precisionByLabel))
print("Training Recall:    " + str(trainingSummary.recallByLabel))
print("Training FMeasure:  " + str(trainingSummary.fMeasureByLabel()))
print("Training AUC:       " + str(trainingSummary.areaUnderROC))

rf_classifier = RandomForestClassifier(
    # numTrees=100,  # Number of decision trees in the forest
    # maxDepth=10,   # Maximum depth of each decision tree
    labelCol="label",
    featuresCol="features"
)

rf_model = rf_classifier.fit(trainingDataTransformed)
trainingSummary = rf_model.summary

print("Training Accuracy:  " + str(trainingSummary.accuracy))
print("Training Precision: " + str(trainingSummary.precisionByLabel))
print("Training Recall:    " + str(trainingSummary.recallByLabel))
print("Training FMeasure:  " + str(trainingSummary.fMeasureByLabel()))
print("Training AUC:       " + str(trainingSummary.areaUnderROC))

gbt = GBTClassifier(
    featuresCol="features", 
    labelCol="label")
gbt_model = gbt.fit(trainingDataTransformed)

binary_evaluator = BinaryClassificationEvaluator(labelCol="label", metricName="areaUnderROC")
predictions = gbt_model.transform(trainingDataTransformed)

auc = binary_evaluator.evaluate(predictions)
print(auc)

# print("Training Accuracy:  " + str(trainingSummary.accuracy))
# print("Training Precision: " + str(trainingSummary.precisionByLabel))
# print("Training Recall:    " + str(trainingSummary.recallByLabel))
# print("Training FMeasure:  " + str(trainingSummary.fMeasureByLabel()))
# print("Training AUC:       " + str(trainingSummary.areaUnderROC))

# Load in the tables
test_df = spark.sql("select * from default.reviews_test")
test_df.show(5)
print((test_df.count(), len(test_df.columns)))

test_df = test_df.fillna("None", subset=['reviewText', 'summary'])

def freq_word_count_test(data):

    regexTokenizer = RegexTokenizer(inputCol="reviewText", outputCol="words", pattern="\\W")
    data = regexTokenizer.transform(data)

    data = data.withColumn("wordCount", size(data["words"]))

    freq_word_list = [row.word for row in filtered_words.select("word").collect()]
    freq_word_set = set(freq_word_list)

    def count_freq_words(words):
        return len([word for word in words if word in freq_word_set])

    count_freq_words_udf = udf(count_freq_words, IntegerType())
    data = data.withColumn("freq_word_count", count_freq_words_udf(col("words")))

    return data

drop_list = [
    'asin', 
    # 'reviewID', 
    'reviewerID', 
    'unixReviewTime',
    'reviewTime', 
    'image', 
    'style', 
    'reviewerName', 
    'overall', 
    'stemmed_text', 
    'lowered_text', 
    'cleaned_text', 
    'words', 
    'filtered_text']

test_df = test_df.select([column for column in test_df.columns if column not in drop_list])
print((test_df.count(), len(test_df.columns)))